package produce.controller;

import com.alibaba.fastjson.JSON;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import produce.config.MySource;
import produce.model.Args;
import produce.model.User;

import java.util.Random;

/**
 * @author xiaohong
 * @version 1.0
 * @date 2021/12/15 0015 12:26
 * @description 发送普通消息
 */
@RestController
@RequestMapping("/normal")
@SuppressWarnings("all")
@Slf4j
public class SendController {

    @Autowired
    private MySource mySource;//<1>

    /**
     * 集群消费
     * @return
     */
    @GetMapping("/send_cluster")
    public boolean sendCluster() {
        // <2>创建 Message
        User message = new User();
        message.setId(new Random().nextInt());
        // <3>创建 Spring Message 对象
        Message<User> springMessage = MessageBuilder.withPayload(message)
                .build();
        log.info("[集群消费消息][发送消息完成, 结果 = {}]", springMessage);
        // <4>发送消息 可以设置第二个超时时间参数 默认是-1
        return mySource.clusterOutput().send(springMessage);
    }

    /**
     * 广播消费
     * @return
     */
    @GetMapping("/send_radio")
    public boolean sendRadio() {
        // <2>创建 Message
        User message = new User();
        message.setId(new Random().nextInt());
        // <3>创建 Spring Message 对象
        Message<User> springMessage = MessageBuilder.withPayload(message)
                .build();
        log.info("[广播消费消息][发送消息完成, 结果 = {}]", springMessage);
        // <4>发送消息
        return mySource.radioOutput().send(springMessage);
    }

    /**
     * 延迟消息
     延迟级别	时间	延迟级别	时间	延迟级别	时间
     1	        1s	    7	3m	13	    9m
     2	        5s	    8	4m	14	    10m
     3	        10s 	9	5m	15	    20m
     4	        30s	    10	6m	16	    30m
     5	        1m	    11	7m	17	    1h
     6	        2m	    12	8m	18	    2h
     * @return
     */
    @GetMapping("/send_delay")
    public boolean sendDelay() {
        // 创建 Message
        User message = new User();
        message.setId(new Random().nextInt());
        // 创建 Spring Message 对象
        Message<User> springMessage = MessageBuilder.withPayload(message)
                .setHeader(MessageConst.PROPERTY_DELAY_TIME_LEVEL, "3") // 设置延迟级别为 3，10 秒后消费。
                .build();
        // 发送消息
        boolean sendResult = mySource.clusterOutput().send(springMessage);
        log.info("[延迟消费消息][发送消息完成, 结果 = {}]", springMessage);
        return sendResult;
    }


    /**
     * 消息重试
     * 默认从定时消息的第三级别开始重试，一直到18级，默认重试16次
     * 只有集群消费模式下，才有消息重试
     *
     * @return
     */
    @GetMapping("/send_retry")
    public boolean sendReTry() {
        // 创建 Message
        User message = new User();
        message.setId(new Random().nextInt());
        // <3>创建 Spring Message 对象
        Message<User> springMessage = MessageBuilder.withPayload(message)
                .build();
        // 发送消息
        boolean sendResult = mySource.retryOutput().send(springMessage);
        log.info("[消息重试消息][发送消息完成, 结果 = {}]", springMessage);
        return sendResult;
    }

    /**
     * 顺序消息
     * 普通顺序消息：Producer 将相关联的消息发送到相同的消息队列。
     * 完全严格顺序：在【普通顺序消息】的基础上，Consumer 严格顺序消费。
     * @return
     */
    @GetMapping("/send_order")
    public boolean sendOrder() {
        // 发送 3 条相同 id 的消息
        int id = new Random().nextInt();
        for (int i = 0; i < 3; i++) {
            // 创建 Message
            User message = new User();
            message.setId(id);
            // 创建 Spring Message 对象
            Message<User> springMessage = MessageBuilder.withPayload(message)
                    .build();
            // 发送消息
            mySource.orderOutput().send(springMessage);
        }
        log.info("[顺序消息][发送消息完成, 条数 = {}]", 3);
        return true;
    }

    /**
     * 根据tag过滤消息
     * @return
     */
    @GetMapping("/send_filter")
    public boolean sendFilter() {
        for (String tag : new String[]{"yunai", "yutou", "tudou"}) {
            // 创建 Message
            User message = new User();
            message.setId(new Random().nextInt());
            // 创建 Spring Message 对象
            Message<User> springMessage = MessageBuilder.withPayload(message)
                    .setHeader(MessageConst.PROPERTY_TAGS, tag) // 设置 Tag
                    .build();
            // 发送消息
            mySource.filterOutput().send(springMessage);
        }
        log.info("[根据tag过滤消息][发送消息完成, 条件 = {}]", new String[]{"yunai", "yutou", "tudou"});
        return true;
    }


    /**
     * 测试事务消息
     * @return
     */
    @GetMapping("/send_transactional")
    public boolean sendTransactional() {
        // 创建 Message
        User message = new User();
        message.setId(new Random().nextInt());
        // 创建 Spring Message 对象
        Args args = new Args();
        args.setArgs1(1);
        args.setArgs2("2");
        Message<User> springMessage = MessageBuilder.withPayload(message)
                .setHeader("args", JSON.toJSONString(args)) // <X>
                .build();
        // 发送消息
        return mySource.transactionalOutput().send(springMessage);
    }



}
